package drds.binlog.server.netty.handler;

import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.zookeeper.running.Monitor;
import drds.binlog.common.zookeeper.running.ServerRunningMonitors;
import drds.binlog.protocol.protocol_proto.ClientAuth;
import drds.binlog.protocol.protocol_proto.Packet;
import drds.binlog.server.embedded.EmbeddedServer;
import drds.binlog.server.netty.Nettys;
import org.apache.commons.lang.StringUtils;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.*;
import org.jboss.netty.handler.timeout.IdleStateAwareChannelHandler;
import org.jboss.netty.handler.timeout.IdleStateEvent;
import org.jboss.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.util.concurrent.TimeUnit;

/**
 * 客户端身份认证处理
 */
public class ClientAuthenticationHandler extends SimpleChannelHandler
{

    private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class);
    private final int SUPPORTED_VERSION = 3;
    private final int defaultSubscriptorDisconnectIdleTimeout = 60 * 60 * 1000;
    private EmbeddedServer embeddedServer;

    public ClientAuthenticationHandler()
    {

    }

    public ClientAuthenticationHandler(EmbeddedServer embeddedServer)
    {
        this.embeddedServer = embeddedServer;
    }

    public void messageReceived(final ChannelHandlerContext channelHandlerContext, MessageEvent messageEvent) throws Exception
    {
        ChannelBuffer channelBuffer = (ChannelBuffer) messageEvent.getMessage();
        final Packet packet = Packet.parseFrom(channelBuffer.readBytes(channelBuffer.readableBytes()).array());
        switch (packet.getVersion())
        {
            case SUPPORTED_VERSION:
            default:
                final ClientAuth clientAuth = ClientAuth.parseFrom(packet.getBody());
                // 如果存在订阅信息
                if (StringUtils.isNotEmpty(clientAuth.getDestination())
                        && StringUtils.isNotEmpty(clientAuth.getClientId()))
                {
                    ClientInfo clientInfo = new ClientInfo(clientAuth.getDestination(),
                            Short.valueOf(clientAuth.getClientId()),
                            clientAuth.getFilter());
                    try
                    {
                        MDC.put("destination", clientInfo.getDestination());
                        embeddedServer.subscribe(clientInfo);
                        channelHandlerContext.setAttachment(clientInfo);// 设置状态数据
                        // 尝试启动，如果已经启动，忽略
                        if (!embeddedServer.isStart(clientInfo.getDestination()))
                        {
                            Monitor monitor = ServerRunningMonitors.getRunningMonitor(clientInfo.getDestination());
                            if (!monitor.isRunning())
                            {
                                monitor.start();
                            }
                        }
                    } finally
                    {
                        MDC.remove("destination");
                    }
                }
                // 鉴权一次性，暂不统计
                Nettys.ack(channelHandlerContext.getChannel(), new ChannelFutureListener()
                {

                    public void operationComplete(ChannelFuture channelFuture) throws Exception
                    {
                        logger.info("remove unused channel handlers after authentication is done successfully.");
                        channelHandlerContext.getPipeline().remove(HandshakeInitializationHandler.class.getName());
                        channelHandlerContext.getPipeline().remove(ClientAuthenticationHandler.class.getName());

                        int readTimeout = defaultSubscriptorDisconnectIdleTimeout;
                        int writeTimeout = defaultSubscriptorDisconnectIdleTimeout;
                        if (clientAuth.getNetReadTimeout() > 0)
                        {
                            readTimeout = clientAuth.getNetReadTimeout();
                        }
                        if (clientAuth.getNetWriteTimeout() > 0)
                        {
                            writeTimeout = clientAuth.getNetWriteTimeout();
                        }
                        // fix bug: soTimeout parameter's unit from connector is
                        // millseconds.
                        IdleStateHandler idleStateHandler = new IdleStateHandler(Nettys.hashedWheelTimer,
                                readTimeout,
                                writeTimeout,
                                0,
                                TimeUnit.MILLISECONDS);
                        channelHandlerContext.getPipeline().addBefore(SessionHandler.class.getName(),
                                IdleStateHandler.class.getName(),
                                idleStateHandler);

                        IdleStateAwareChannelHandler idleStateAwareChannelHandler = new IdleStateAwareChannelHandler()
                        {

                            public void channelIdle(ChannelHandlerContext channelHandlerContext1, IdleStateEvent e) throws Exception
                            {
                                logger.warn("channel:{} idle timeout exceeds, close channel to save server resources...",
                                        channelHandlerContext1.getChannel());
                                channelHandlerContext1.getChannel().close();
                            }

                        };
                        channelHandlerContext.getPipeline().addBefore(SessionHandler.class.getName(),
                                IdleStateAwareChannelHandler.class.getName(),
                                idleStateAwareChannelHandler);
                    }

                });
                break;
        }
    }

    public void setEmbeddedServer(EmbeddedServer embeddedServer)
    {
        this.embeddedServer = embeddedServer;
    }

}
